package com.example.mysql.zendesk.framework;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class BinlogDispatcherListener implements BinaryLogClient.EventListener {
    private static final Logger log = LoggerFactory.getLogger(BinlogDispatcherListener.class);

    private final Map<String, TableListener> tableListenerMap = new HashMap<>();
    private final Map<Long, String> tableMapping = new ConcurrentHashMap<>(64);

    public BinlogDispatcherListener(Set<TableListener> listeners) {
        listeners.forEach(i -> tableListenerMap.put((i.database() + "." + i.table()).toUpperCase(), i));
    }

    @Override
    public void onEvent(Event event) {
        EventHeader header = event.getHeader();
        EventType eventType = header.getEventType();
        log.debug("eventType: {}", eventType);
        if (event.getData() instanceof TableMapEventData data) {
            tableMapping.put(data.getTableId(), (data.getDatabase() + "." + data.getTable()).toUpperCase());
            return;
        }
        // 仅支持 insert, update, delete
        if (!EventType.isWrite(eventType) && !EventType.isUpdate(eventType) && !EventType.isDelete(eventType)) {
            return;
        }
        if (EventType.isWrite(eventType)) {
            WriteRowsEventData data = event.getData();
            log.debug("write: {}", data);
            this.getListener(data.getTableId()).ifPresent(i -> i.onWrite(eventType, data));
        } else if (EventType.isUpdate(eventType)) {
            UpdateRowsEventData data = event.getData();
            log.debug("update: {}", data);
            this.getListener(data.getTableId()).ifPresent(i -> i.onUpdate(eventType, data));
        } else if (EventType.isDelete(eventType)) {
            DeleteRowsEventData data = event.getData();
            log.debug("delete: {}", data);
            this.getListener(data.getTableId()).ifPresent(i -> i.onDelete(eventType, data));
        }
    }

    private Optional<TableListener> getListener(long tableId) {
        String name = tableMapping.get(tableId);
        return StringUtils.hasLength(name) ? Optional.ofNullable(tableListenerMap.get(name)) : Optional.empty();
    }
}
